package com.cnpc.bds.sdk.sink.kafka.executor;

import com.cnpc.bds.sdk.sink.kafka.handler.Pipeline;

import java.util.LinkedList;
import java.util.List;

/**
 * 消息事件广播器.
 * <p>
 * 上接Kafka消费任务，下承消息责任链处理.
 * </p>
 *
 * @author ssp
 * @since 1.0
 */
public class EventBroadcaster {

    private final List<EventLoop> eventLoops = new LinkedList<>();

    private EventLoopFactory eventLoopFactory;

    public EventBroadcaster(String downEventLoopGroupName) {
        this.eventLoopFactory = new EventLoopFactory(downEventLoopGroupName);
    }

    public void broadcast(Object msg) {
        //消息广播
        for (EventLoop eventLoop : eventLoops) {
            eventLoop.fireEventAsync(msg); //异步提交
        }
    }

    public void join(EventLoop eventLoop) {
        eventLoops.add(eventLoop);
    }

    public void join(Pipeline downStreamPipeline) {
        //为pipeline分配EventLoop执行线程.
        EventLoop eventLoop = eventLoopFactory.wrapAndStart(EventLoop.newOne(downStreamPipeline));
        join(eventLoop);
    }

}
